import com.forcht.common.domain.Message;
import com.forcht.common.domain.MessageHeader;
import com.forcht.common.util.ProtostuffUtil;
import com.forcht.common.util.SpringContextUtil;
import com.forcht.server.handler.MessageHandler;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by forcht on 2018/4/11.
 * 服务器
 */
public class Server {

    private static final int DEFAULT_BUFFER_SIZE=1024;
    private static final int DEFAULT_PORT=80;
    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    private ExecutorService threadPool;
    private  MessageHandler messageHandler;

    public Server() {
        init();
    }

    /**
     * 初始化服务器
     *
     */
    private void init() {
        try {
            //创建一个服务器通道
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT));
            //设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            //创建一个选择器
            selector = Selector.open();
            //服务器通道向selector注册，服务器只关心是否准备好接受一个连接
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            //创建一个线程池
            threadPool = Executors.newCachedThreadPool();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动服务器
     */
    public void launch(){
        new ListenerThread().start();
    }

    /**
     * 处理用户请求连接
     */
    private void handleAcceptRequest() {
        try {
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            socketChannel.register(selector, SelectionKey.OP_READ);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 监听通道是否就绪
     */
    private class ListenerThread extends Thread {
        @Override
        public void run() {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    //当注册在selector上的通道有事件到达时，方法返回，否则select()方法阻塞
                    selector.select();
                    //遍历所有的事件
                    Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                    while (keys.hasNext()) {
                        //获得监听的事件
                        SelectionKey key = keys.next();
                        //删除该事件
                        keys.remove();
                        //如果是请求连接的事件
                        if (key.isAcceptable()) {
                            handleAcceptRequest();
                        } else if (key.isReadable()) {
                            //如果是请求读取的事件
                            //取消监听可读事件，本次处理完后才能打开
                            key.interestOps(key.interestOps()&~SelectionKey.OP_READ);
                            //提交给线程池处理
                            threadPool.execute(new ReadEventHandler(key));
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 读取数据
     */
    private class ReadEventHandler implements Runnable {

        private SocketChannel socketChannel;
        private ByteBuffer buffer;
        private SelectionKey key;
        private ByteArrayOutputStream out;
        public ReadEventHandler(SelectionKey key) {
            this.key = key;
            socketChannel = (SocketChannel) key.channel();
            buffer = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
            out = new ByteArrayOutputStream();
        }

        @Override
        public void run() {
            try {
                //读取数据
                int size = 0;
                while ((size = socketChannel.read(buffer))>0) {
                    buffer.flip();
                    out.write(buffer.array(), 0, size);
                    buffer.clear();
                }
                //如果客户端主动关闭连接(即调用close方法)服务器会监听到可读事件
                //并且read会返回-1
                if(size==-1){
                    System.out.println("客户端主动关闭连接");
                    closeClientSocket();
                    return ;
                }
                //数据读取结束，继续监听读事件
                key.interestOps(key.interestOps()|SelectionKey.OP_READ);
                key.selector().wakeup();
                //获取读取的字节数组
                byte[] bytes = out.toByteArray();
                out.close();
                //反序列化对象
                Message message = ProtostuffUtil.deserialize(bytes, Message.class);
                //获取消息处理器
                MessageHeader header = message.getHeader();
                messageHandler= SpringContextUtil.getBean("messageHandler",header.getMessageType().toString().toLowerCase());
                //提交给消息处理器处理
                messageHandler.handle(message,key);
            } catch (IOException e) {
                //客户端强制/异常关闭(即没有调用close()方法关闭时)，服务器会监听到可读事件
                //服务器进行读取数据时，由于连接已关闭，会抛出远程主机强制关闭连接的异常
                System.out.println("客户端没有安全关闭");
                closeClientSocket();
            }
        }

        /**
         * 关闭客户端的连接
         */
        private void closeClientSocket(){
            try {
                key.cancel();
                socketChannel.socket().close();
                socketChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

    public static void main(String[] args) {
        Server server=new Server();
        server.launch();
    }
}
